package com.gin.monitor.agg.warn;

import com.gin.monitor.agg.utils.JobEnvUtils;
import com.gin.monitor.agg.vo.PoliceAction;
import com.gin.monitor.agg.vo.ViolationInfo;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.ProcessJoinFunction;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.Collector;

import java.time.Duration;


/**
 * 违法车辆和交警出警分析
 * 第一种，当前的违法车辆（在5分钟内）如果已经出警了。（最后输出道主流中做删除处理）。
 * 第二种，当前违法车辆（在5分钟后）交警没有出警（发出出警的提示，在侧流中发出）。
 * 第三种，有交警的出警记录，但是不是由监控平台报的警。
 * 需要两种数据流：
 * 1、系统的实时违法车辆的数据流
 * 2、交警实时出警记录数据
 * <p>
 * <p>
 * 测试时注意: 使用当前时间戳
 * nc -lk 9999
京L22189,套牌,1620637884000
京L22188,套牌,1620637885000
京L22187,套牌,1620637886000
京L22187,套牌,1620637996000
 * <p>
 * nc -lk 8888
007,京L22187,出警,1620637885000
008,京L22188,处理完成,1620637886000
 *
 * @author gin
 * @date 2021/4/30
 */
public class ViolationCarAndPoliceActionAnalysis {

    public static void main(String[] args) {
        StreamExecutionEnvironment env = JobEnvUtils.initEnv(args);

        //第一种，当前的违法车辆（在5分钟内）如果已经出警了。（最后输出道主流中做删除处理）。
        // nc -lk 9999
        SingleOutputStreamOperator<ViolationInfo> stream1 = env.socketTextStream("node01", 9999)
                .map(new RichMapFunction<String, ViolationInfo>() {
                    @Override
                    public ViolationInfo map(String line) {
                        String[] s = line.split(",");
                        return new ViolationInfo(s[0], s[1], Long.parseLong(s[2]));
                    }
                })
                .assignTimestampsAndWatermarks(WatermarkStrategy
                        .<ViolationInfo>forBoundedOutOfOrderness(Duration.ofSeconds(2))
                        .withTimestampAssigner((event, timestamp) -> event.createTime));

        // nc -lk 8888
        SingleOutputStreamOperator<PoliceAction> stream2 = env.socketTextStream("node01", 8888)
                .map(new RichMapFunction<String, PoliceAction>() {
                    @Override
                    public PoliceAction map(String line) {
                        String[] s = line.split(",");
                        return new PoliceAction(s[0], s[1], s[2], Long.parseLong(s[3]));
                    }
                })
                .assignTimestampsAndWatermarks(WatermarkStrategy
                        .<PoliceAction>forBoundedOutOfOrderness(Duration.ofSeconds(2))
                        .withTimestampAssigner((event, timestamp) -> event.actionTime));

        //需要两个数据流的连接，而且这两个中数据流有连接条件：PoliceAction.car =ViolationInfo.car
        //这个连接相当于数据库中的内连接
        stream1.keyBy((KeySelector<ViolationInfo, String>) violationInfo -> violationInfo.car)
                //内连接 intervalJoin只能应用在keyedStream，普通的DataStream流连接只能使用connect
                .intervalJoin(stream2.keyBy((KeySelector<PoliceAction, String>) policeAction -> policeAction.car))
                // 设置一个时间边界，在这个边界中，两个流的数据自动关联; 超出时间边界则不会关联命中
                // 为了便于测试，使用30秒替代5分钟 ,和Watermark的延迟没有关系
                .between(Time.seconds(-30), Time.seconds(30))
                .process(new ProcessJoinFunction<ViolationInfo, PoliceAction, String>() {
                    @Override
                    public void processElement(ViolationInfo left, PoliceAction right, Context ctx, Collector<String> out) {
                        out.collect("车辆" + left.car + ",已经有交警出警了，警号为:" + right.policeId + ",出警的状态是：" +
                                right.actionStatus + ",出警的时间:" + right.actionTime);
                    }
                })
                .print();


        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }

    }

}
